package com.xctech.yace.service;

import com.xctech.yace.enumeration.KafkaRoleType;
import com.xctech.yace.pool.ProduceThreadPool;
import com.xctech.yace.thread.ConsumerThread;
import com.xctech.yace.thread.ConsumerTpsThread;
import com.xctech.yace.util.DirUtil;
import com.xctech.yace.util.KafkaKerberosUtil;
import lombok.Data;
import org.apache.kafka.clients.consumer.KafkaConsumer;


import java.io.FileInputStream;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ThreadPoolExecutor;

import static com.xctech.yace.util.KafkaKerberosUtil.getKafkaSaslPlainConfMap;


@Data
public class KafkaRecived {

    private Properties props;

    private KafkaConsumer<String,String> consumer;

    private Integer partitionNum = 0;

    public KafkaRecived() {
        String filePath = DirUtil.getOrgPath()+"conf/kafkaconsumer.properties";
        props = new Properties();
        Properties newProps = new Properties();
        try {
            props.load(new FileInputStream(filePath));
            Map<String, String> map = getKafkaSaslPlainConfMap(props, KafkaRoleType.CONSUMER);
            if(map != null){
                for(Map.Entry<String,String> entry:map.entrySet()){
                    String key = entry.getKey();
                    String value = entry.getValue();
                    props.setProperty(key,value);
                }
            }else{
                String mechanism = KafkaKerberosUtil.mechanism;
                String protocol = KafkaKerberosUtil.protocol;
                for (String field:props.stringPropertyNames()){
                    if (!(mechanism.equals(field) || protocol.equals(field))) {
                        newProps.setProperty(field,props.getProperty(field,""));
                    }
                }
                props = newProps;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.consumer = new KafkaConsumer<String, String>(props);

//        String ps = (String) props.get("partition.num");
//        this.partitionNum = Integer.parseInt(ps.trim());

    }

    public static void main(String[] args) {
        KafkaRecived recived = new KafkaRecived();
        String[] topics = new String[]{"CDC"};
        recived.reciveMsg(topics);
    }
    /**
     * 开启接受消息
     * @param topics 主题列表
     * @return
     */
    public ConsumerThread reciveMsg(String[] topics){
        ConsumerThread consumerThread = new ConsumerThread(consumer, topics,0);
        ThreadPoolExecutor instance = ProduceThreadPool.getInstance();
        instance.execute(consumerThread);
        return consumerThread;
    }

    public List<ConsumerThread> reciveConsumer(String[] topics){
        List consumerList = new ArrayList<ConsumerThread>();
        for(int i =0 ;i< this.partitionNum; i++){
            KafkaConsumer newconsumer = new KafkaConsumer<String, String>(props);
            ConsumerThread consumerThread = new ConsumerThread(newconsumer, topics,i);
            consumerList.add(consumerThread);
        }
        return consumerList;
    }

    /**
     * 开启接受消息tps
     * @param topics 主题列表
     * @return
     */
    public void reciveMsgTps(String[] topics){
        ProduceThreadPool.getInstance().execute(new ConsumerTpsThread(consumer,topics));
    }

    /**
     * 线程池停止
     * @return
     */
    public void stopReciveThreadPool(Integer second){
        ProduceThreadPool.stopPool(1000l*second);
    }

}
